package com.stxx.stiot2.event;

import com.stxx.framwork.constants.AlarmPolicyType;
import com.stxx.framwork.constants.ChangeType;
import com.stxx.framwork.constants.StorePolicyType;
import com.stxx.framwork.utils.FluxUtils;
import com.stxx.framwork.utils.ObjectUtils;
import com.stxx.stiot2.constants.RedisKeys;
import com.stxx.stiot2.domain.Gateway;
import com.stxx.stiot2.domain.Scene;
import com.stxx.stiot2.domain.Terminal;
import com.stxx.stiot2.domain.cache.GatewayCache;
import com.stxx.stiot2.domain.cache.TerminalCache;
import com.stxx.stiot2.domain.ext.*;
import com.stxx.stiot2.event.dto.T2pMsg;
import com.stxx.stiot2.event.task.SceneTask;
import com.stxx.stiot2.event.task.SubscribeReportTask;
import com.stxx.stiot2.external.InfluxdbServer;
import com.stxx.stiot2.external.RedisServer;
import com.stxx.stiot2.service.*;
import com.stxx.stiot2.utils.SignalUtils;
import com.stxx.stiot2.ws.WsController;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.noear.solon.Solon;
import org.noear.solon.annotation.Component;
import org.noear.solon.core.event.EventListener;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.SQLException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;


/**
 * @author Fuzhenlei
 * @date 2024/12/17 17:48
 */
@Slf4j
@AllArgsConstructor
@Component
public class ReportListener implements EventListener<T2pMsg> {
    private final InfluxdbServer influxdbServer;
    private final TerminalService terminalService;
    private final ProfileService profileService;
    private final SceneService sceneService;
    //private RedisServer redisServer;
    private final Executor executor;
    private final GatewayService gatewayService;
    @Override
    public void onEvent(T2pMsg message) {
        long ts = System.currentTimeMillis();
        GatewayCache gatewayCache = gatewayService.getCache(message.getGatewayCode());
        TerminalCache terminalCache =  terminalService.getCache(message.getTerminalCode());
        if(gatewayCache==null){
            log.error("gateway "+message.getGatewayCode()+" is not fount");
            return;
        }
        try {
            //设置网关上报时间
            if(gatewayCache.getLastReportTime()==null||ts - gatewayCache.getLastReportTime()>60000){
                gatewayCache.setLastReportTime(ts);
                gatewayService.updateLastReportTime(gatewayCache);
            }
            if(!gatewayCache.getRegiTerminal()&&terminalCache==null){
                log.error("terminal "+message.getTerminalCode()+" is not fount");
                return;
            }
            //设置终端最后上报时间
            if(terminalCache!=null){
                if(terminalCache.getLastReportTime()==null||ts - terminalCache.getLastReportTime()>60000){
                    terminalCache.setLastReportTime(ts);
                    terminalService.updateLastReportTime(terminalCache);
                }
            }

        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

        ReportData reportData = message.getData();
        String pr = reportData.getProfile();
        Map<String, SignalEx> signals = profileService.signals(reportData.getProfile());
        if(signals==null){
         throw new RuntimeException("没有找到终端配置");
        }
        //Terminal terminal = terminalService.queryById(message.getTerminalCode());
        if(terminalCache == null) {
            Terminal terminal = new Terminal();
            terminal.setCode(message.getTerminalCode());
            terminal.setName(message.getTerminalCode());
            terminal.setGatewayCode(message.getGatewayCode());
            terminal.setProfileCode(reportData.getProfile());
            try {
                terminalService.insert(terminal);
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }



        Map<String,Object> saveData = new HashMap<>();
        //Map<String,Object> pushData = new HashMap<>();
        Map<String,String> alarmInfo = new HashMap<>();
        //处理数据点
        Map<String, Object> lastValues = terminalService.getLastValues(terminalCache.getCode());
        reportData.getProperties().forEach((k, v) -> {
            if(!signals.containsKey(k))return;
            SignalEx signal = signals.get(k);
            Class<?> fieldType = FluxUtils.getType(signal.getFieldType());
            //当前值
            Object curValue = ObjectUtils.get(v, fieldType);
            reportData.getProperties().put(k,curValue);
            //pushData.put(k,curValue);
            //上次值
            Object oldValue = lastValues.get(signal.getCode());
            //变化存储
            if(signal.getStorePolicy()== StorePolicyType.NEVER){
                return;
            }else if(signal.getStorePolicy()== StorePolicyType.CHANGE){
                if(!SignalUtils.isChange(oldValue,curValue,signal))return;
            }else if(signal.getStorePolicy()== StorePolicyType.PERIOD){
                //if(!SignalUtils.isTime(signal,ts))return;
                return;
            }else if(signal.getStorePolicy()== StorePolicyType.PER_CH){
                if(!SignalUtils.isChange(oldValue,curValue,signal)){
                    return;
                }
            }
            saveData.put(k,curValue);
        });
        terminalService.setLastValues(terminalCache.getCode(),reportData.getProperties());
        //执行联动
        List<Scene> scenes = sceneService.queryForExec(terminalCache.getProfileCode(), terminalCache.getCode());
        if(scenes!=null&&scenes.size()>0){
            scenes.forEach(item->{
                executor.execute(new SceneTask(item,message));
            });
        }
        //推送数据
        WsController.push(terminalCache.getCode(),reportData.getProperties(),alarmInfo);
        //推送到订阅
        executor.execute(new SubscribeReportTask(message));


        if(saveData.isEmpty()){
            return;
        }
        //写入influxDb
        influxdbServer.write(terminalCache.getProfileCode(),terminalCache.getGatewayCode(),terminalCache.getCode(),null,saveData,reportData.getTimestamp());

    }


//    public static void main(String[] args) {
//        BigDecimal old = new BigDecimal(12);
//        BigDecimal cur = new BigDecimal(10);
//        if(old.compareTo(BigDecimal.valueOf(0))!=0){
//            if(old.subtract(cur)
//                    .divide(old,5, RoundingMode.HALF_UP).multiply(BigDecimal.valueOf(100))
//                    .abs().compareTo(BigDecimal.valueOf(1))>0){
//                System.out.println("ok");
//                return;
//            }
//        }
//    }
}
